package com.basic.rxjava3.lession;

import io.reactivex.rxjava3.annotations.NonNull;
import io.reactivex.rxjava3.core.*;
import io.reactivex.rxjava3.disposables.CompositeDisposable;
import io.reactivex.rxjava3.disposables.Disposable;
import io.reactivex.rxjava3.functions.Action;
import io.reactivex.rxjava3.functions.Consumer;
import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.Test;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

@Slf4j
public class RxJava3Lession1 {

    /*
     * Observeable用于订阅Observer，是不支持背压的
     * 这种观察者模型不支持背压：当被观察者快速发送大量数据时，下游不会做其他处理，即使数据大量堆积，调用链也不会报MissingBackpressureException,
     * 消耗内存过大只会OOM。
     * 所以，当我们使用Observable/Observer的时候，我们需要考虑的是，数据量是不是很大(官方给出以1000个事件为分界线作为参考)
     * */
    @Test
    public void test1() {
        Observable mObservable = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                emitter.onNext(1);
                Thread.sleep(500);
                emitter.onNext(2);
                Thread.sleep(800);
                emitter.onComplete();
            }
        });

        Observer mObserver = new Observer<Integer>() {
            //这是新加入的方法，在订阅后发送数据之前，
            //回首先调用这个方法，而Disposable可用于取消订阅
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(Integer value) {
                log.info("lucas " + " onNext: " + value);
            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onComplete() {
                log.info("onComplete");
            }
        };
        mObservable.subscribe(mObserver);
    }

    @Test
    public void test2() {
        Flowable.range(0, 10)
                .subscribe(new Subscriber<Integer>() {
                    Subscription sub;

                    //当订阅后，会首先调用这个方法，其实就相当于onStart()，
                    //传入的Subscription s参数可以用于请求数据或者取消订阅
                    @Override
                    public void onSubscribe(Subscription s) {
                        log.info("TAG " + " onsubscribe start");
                        sub = s;
                        sub.request(1);
                        log.info("TAG " + " onsubscribe end");
                    }

                    @Override
                    public void onNext(Integer o) {
                        log.info("TAG " + " onNext--->" + o);
                        sub.request(1);
                    }

                    @Override
                    public void onError(Throwable t) {
                        t.printStackTrace();
                    }

                    @Override
                    public void onComplete() {
                        log.info("TAG " + " onComplete");
                    }
                });
    }


    /*Flowable也可以通过creat()来创建：*/
    @Test
    public void test3() {
        Flowable<Integer> integerFlowable = Flowable.create(new FlowableOnSubscribe<Integer>() {
                                                                @Override
                                                                public void subscribe(FlowableEmitter<Integer> e) throws Exception {
                                                                    e.onNext(1);
                                                                    e.onNext(2);
                                                                    e.onNext(3);
                                                                    e.onNext(4);
                                                                    e.onComplete();
                                                                }
                                                            }
                //需要指定背压策略
                , BackpressureStrategy.BUFFER);
        integerFlowable.subscribe(new Subscriber<Integer>() {
            Subscription sub;

            //当订阅后，会首先调用这个方法，其实就相当于onStart()，
            //传入的Subscription s参数可以用于请求数据或者取消订阅
            @Override
            public void onSubscribe(Subscription s) {
                log.info("TAG " + " onsubscribe start");
                sub = s;
                sub.request(1);
                log.info("TAG " + " onsubscribe end");
            }

            @Override
            public void onNext(Integer o) {
                log.info("TAG " + " onNext--->" + o);
                sub.request(1);
            }

            @Override
            public void onError(Throwable t) {
                t.printStackTrace();
            }

            @Override
            public void onComplete() {
                log.info("TAG " + " onComplete");
            }
        });
    }

    /*
    * Single类似于Observable，不同的是，它总是只发射一个值，或者一个错误通知，而不是发射一系列的值（当然就不存在背压问题）
    * 所以当你使用一个单一连续事件流，这样你可以使用Single。
    * Single观察者只包含两个事件，一个是正常处理成功的onSuccess，另一个是处理失败的onError。
    * 因此，不同于Observable需要三个方法onNext, onError, onCompleted，订阅Single只需要两个方法：
    onSuccess - Single发射单个的值到这个方法
    onError - 如果无法发射需要的值，Single发射一个Throwable对象到这个方法
    *
    * */
    @Test
    public void test4() {
        //被观察者
        Single<String> single = Single.create(new SingleOnSubscribe<String>() {
            @Override
            public void subscribe(SingleEmitter<String> e) throws Exception {
                e.onSuccess("test");
                e.onSuccess("test2");//错误写法，重复调用也不会处理，因为只会调用一次
            }
        });

        //订阅观察者SingleObserver
        single.subscribe(new SingleObserver<String>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onSuccess(String s) {
                //相当于onNext和onComplete
                log.info(s);
            }

            @Override
            public void onError(Throwable e) {

            }
        });

    }

    /*
     * 如果你的观察者连onNext事件都不关心，可以使用Completable，它只有onComplete和onError两个事件：
     * */
    @Test
    public void test5() {

        Completable.create(new CompletableOnSubscribe() {//被观察者

            @Override
            public void subscribe(CompletableEmitter e) throws Exception {
                e.onComplete();//单一onComplete或者onError
            }

        }).subscribe(new CompletableObserver() {//观察者
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onComplete() {
                log.info("lucas " + " onComplete: ");
            }

            @Override
            public void onError(Throwable e) {

            }
        });

    }


    /*
    如果你有一个需求是可能发送一个数据或者不会发送任何数据，这时候你就需要Maybe，它类似于Single和Completable的混合体。
    Maybe可能会调用以下其中一种情况（也就是所谓的Maybe）：
        onSuccess或者onError
        onComplete或者onError
        可以看到onSuccess和onComplete是互斥的存在
    * */
    @Test
    public void test6() {
        //被观察者
        Maybe<String> maybe = Maybe.create(new MaybeOnSubscribe<String>() {
            @Override
            public void subscribe(MaybeEmitter<String> e) throws Exception {
                e.onSuccess("test");//发送一个数据的情况，或者onError，不需要再调用onComplete(调用了也不会触发onComplete回调方法)
                //e.onComplete();//不需要发送数据的情况，或者onError
            }
        });

        //订阅观察者
        maybe.subscribe(new MaybeObserver<String>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onSuccess(String s) {
                //发送一个数据时，相当于onNext和onComplete，但不会触发另一个方法onComplete
                log.info("lucas " + s);
            }

            @Override
            public void onComplete() {
                //无数据发送时候的onComplete事件
                log.info("lucas " + " onComplete");
            }

            @Override
            public void onError(Throwable e) {

            }
        });

    }

    @Test
    public void test7() {
        CompositeDisposable mRxEvent = new CompositeDisposable();
        Disposable subscribe = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<String> e) throws Exception {
                e.onNext("俊俊俊很帅");
                e.onNext("你值得拥有");
                e.onNext("取消关注");
                e.onNext("但还是要保持微笑");
                e.onComplete();
            }
        }).subscribe(
                new Consumer<String>() {
                    @Override
                    public void accept(@NonNull String s) throws Exception {
                        //对应onNext()
                        System.out.println("accept=" + s);
                    }
                }, new Consumer<Throwable>() {
                    @Override
                    public void accept(@NonNull Throwable throwable) throws Exception {
                        //对应onError()
                    }
                }, new Action() {
                    @Override
                    public void run() throws Exception {
                        //对应onComplete()
                    }
                }, new CompositeDisposable());

        mRxEvent.add(subscribe);
        mRxEvent.clear();
    }


//    @Test
//    public void test2(){}


}